﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Runtime.Loader;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

using Microsoft.Extensions.DependencyModel;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Options;

using Newtonsoft.Json;

using StackExchange.Redis;

namespace Jack.RedisEventBus
{
    internal class SubscriperHostedService : BackgroundService
    {
        private readonly IOptions<EventBusOption> _options;
        private readonly IEventStore _eventStore;
        private readonly string _eventChannel;
        private readonly JsonSerializerSettings _jsonSerializerSettings;
        private ConnectionMultiplexer _connection;

        public SubscriperHostedService(IOptions<EventBusOption> options, IEventStore eventStore)
        {
            _options = options;
            _eventStore = eventStore;
            _eventChannel = $"{options.Value.InstanceName}{options.Value.EventChannel}";
            _jsonSerializerSettings = new JsonSerializerSettings
            {
                //DateFormatHandling = DateFormatHandling.IsoDateFormat,
                //DateParseHandling = DateParseHandling.DateTime,
                DefaultValueHandling = DefaultValueHandling.Ignore,
                //FloatFormatHandling = FloatFormatHandling.String,
                NullValueHandling = NullValueHandling.Ignore,
                DateFormatString = "yyyy'-'MM'-'dd'T'HH':'mm':'ss.fff",
                //MissingMemberHandling = MissingMemberHandling.Ignore,
                //StringEscapeHandling = StringEscapeHandling.Default,
            };
        }

        public override async Task StartAsync(CancellationToken cancellationToken)
        {
            #region 扫描载入事件处理者
            var libs = DependencyContext.Default.CompileLibraries.Where(lib => lib.Serviceable == false && lib.Type == "project").Select(lib => lib.Name).ToList();
            var assemblies = libs.Select(lib => AssemblyLoadContext.Default.LoadFromAssemblyName(new System.Reflection.AssemblyName(lib))).ToArray();

            foreach (var assemblie in assemblies)
            {
                foreach (var eventHandlerType in assemblie.GetExportedTypes().Where(x => IsAssignableTo<IEventHandle>(x) && x.IsGenericType == false && x.IsAbstract == false && x.IsClass))
                {
                    var @interface = eventHandlerType.GetInterface("IEventHandle`1");
                    var eventType = @interface.GetGenericArguments()[0];
                    string eventName = eventType.FullName;
                    if (eventType.IsDefined(typeof(EventNameAttribute), false))
                    {
                        eventName = eventType.GetCustomAttribute<EventNameAttribute>().Name;
                    }

                    await _eventStore.AttachEventHandle(eventName, eventHandlerType);
                }
            }
            #endregion

            await base.StartAsync(cancellationToken);
        }

        protected async override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            while (stoppingToken.IsCancellationRequested == false)
            {
                try
                {
                    await Connect(stoppingToken);
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex);
                }
            }
        }

        private async Task Connect(CancellationToken stoppingToken)
        {
            var connection = await ConnectionMultiplexer.ConnectAsync(_options.Value.ConnectionString);
            while (stoppingToken.IsCancellationRequested == false && connection.IsConnected)
            {
                if (await Subscriber(stoppingToken, connection) == false)
                    await Task.Delay(5000);
            }
        }

        private async Task<bool> Subscriber(CancellationToken stoppingToken, ConnectionMultiplexer connection)
        {
            int consumerConcurrent = _options.Value.ConsumerConcurrent;
            if (consumerConcurrent <= 0) consumerConcurrent = 3;

            IDatabase database = await GetConnectedDatabase();
            RedisValue[] redisValues = await database.ListRightPopAsync(_eventChannel, consumerConcurrent);
            if (redisValues == null || redisValues.Length == 0) return false;

            List<Task> tasks = new List<Task>(redisValues.Length);
            for (int i = 0; i < redisValues.Length; i++)
            {
                RedisValue redisValue = redisValues[i];
                if (redisValue == RedisValue.Null) continue;

                EventData eventData = (EventData)(string)redisValue;
                tasks.Add(HandlerEvent(eventData));
            }
            await Task.WhenAll(tasks);
            return true;
        }

        private async Task HandlerEvent(EventData eventData)
        {
            string eventType = eventData.EventType;

            try
            {
                var eventHandles = await _eventStore.GetEventHandles(eventType);
                foreach (var eventHandle in eventHandles)
                {
                    await eventHandle.EventHandle(eventData);
                }
            }
            catch (Exception ex)
            {
                eventData.DeliveryTimes++;
                await Republish(eventData);
                Console.WriteLine(ex);
            }
        }

        private async Task Republish(EventData eventData)
        {
            IDatabase database = await GetConnectedDatabase();
            await database.ListLeftPushAsync(_eventChannel, (string)eventData);
        }

        private async Task<IDatabase> GetConnectedDatabase()
        {
            if (_connection == null || _connection.IsConnected == false)
            {
                _connection = await ConnectionMultiplexer.ConnectAsync(_options.Value.ConnectionString);
            }

            var db = _options.Value.Database;
            return _connection.GetDatabase(db);
        }

        public override Task StopAsync(CancellationToken cancellationToken)
        {
            if (_connection != null && _connection.IsConnected)
            {
                _connection.Dispose();
            }
            return base.StopAsync(cancellationToken);
        }

        static bool IsAssignableTo<T>(Type @this)
        {
            if (@this == null)
            {
                throw new ArgumentNullException("this");
            }

            return typeof(T).IsAssignableFrom(@this);
        }
    }
}
